package com.xxl.job.core.thread;

import java.io.PrintWriter;
import java.io.StringWriter;
import java.util.Collections;
import java.util.Date;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.FutureTask;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.xxl.job.core.biz.model.HandleCallbackParam;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.biz.model.TriggerParam;
import com.xxl.job.core.executor.XxlJobExecutor;
import com.xxl.job.core.handler.IJobHandler;
import com.xxl.job.core.log.XxlJobFileAppender;
import com.xxl.job.core.log.XxlJobLogger;
import com.xxl.job.core.util.ShardingUtil;

/**
 * handler thread 任务执行线程
 * 
 * @author xuxueli 2016-1-16 19:52:47
 */
public class JobThread extends Thread {
	private static Logger logger = LoggerFactory.getLogger(JobThread.class);

	private int jobId;// 任务id
	private IJobHandler handler;// handler
	private LinkedBlockingQueue<TriggerParam> triggerQueue;// 触发器参数队列(调用执行器进行执行)
	private Set<Integer> triggerLogIdSet; // avoid repeat trigger for the same TRIGGER_LOG_ID

	private volatile boolean toStop = false;
	private String stopReason;

	private boolean running = false; // if running job
	private int idleTimes = 0; // idel times

	public JobThread(int jobId, IJobHandler handler) {
		this.jobId = jobId;
		this.handler = handler;
		this.triggerQueue = new LinkedBlockingQueue<TriggerParam>();
		this.triggerLogIdSet = Collections.synchronizedSet(new HashSet<Integer>());
	}

	public IJobHandler getHandler() {
		return handler;
	}

	/**
	 * new trigger to queue 新的触发器增加进来(增加一次要触发执行的任务)
	 * 
	 * @param triggerParam
	 * @return
	 */
	public ReturnT<String> pushTriggerQueue(TriggerParam triggerParam) {
		// avoid repeat
		if (triggerLogIdSet.contains(triggerParam.getLogId())) {
			logger.info(">>>>>>>>>>> repeate trigger job, logId:{}", triggerParam.getLogId());
			return new ReturnT<String>(ReturnT.FAIL_CODE, "repeate trigger job, logId:" + triggerParam.getLogId());
		}

		// 增加要执行的任务的日志id
		triggerLogIdSet.add(triggerParam.getLogId());
		// 增加到触发器参数(增加一次要触发执行的任务)
		triggerQueue.add(triggerParam);
		return ReturnT.SUCCESS;
	}

	/**
	 * kill job thread
	 *
	 * @param stopReason
	 */
	public void toStop(String stopReason) {
		/**
		 * Thread.interrupt只支持终止线程的阻塞状态(wait、join、sleep)， 在阻塞出抛出InterruptedException异常,但是并不会终止运行的线程本身；
		 * 所以需要注意，此处彻底销毁本线程，需要通过共享变量方式；
		 */
		this.toStop = true;
		this.stopReason = stopReason;
	}

	/**
	 * is running job
	 * 
	 * @return
	 */
	public boolean isRunningOrHasQueue() {
		return running || triggerQueue.size() > 0;
	}

	/**
	 * {@inheritDoc}
	 * 
	 * @see java.lang.Thread#run()
	 */
	@Override
	public void run() {

		// init
		try {
			// 执行IJobHandler 中的init方法，以后如果有一些，在执行handler之前的初始化的工作，可以覆写这个方法
			handler.init();
		} catch (Throwable e) {
			logger.error(e.getMessage(), e);
		}

		// execute= stop 为fasle的时候执行
		while (!toStop) {
			running = false;
			// 执行次数
			idleTimes++;

			TriggerParam triggerParam = null;
			ReturnT<String> executeResult = null;
			try {
				// to check toStop signal, we need cycle, so wo cannot use queue.take(), instand of poll(timeout)
				// 从linkBlockingQueue中获取数据，如果3秒获取不到，则返回null
				triggerParam = triggerQueue.poll(3L, TimeUnit.SECONDS);
				if (triggerParam != null) {
					running = true;
					// 将运行次数清空，保证运行90秒空闲之后会被移除
					idleTimes = 0;
					// 去掉这条数据
					triggerLogIdSet.remove(triggerParam.getLogId());

					// log filename, like "logPath/yyyy-MM-dd/9999.log"
					// 创建日志
					String logFileName = XxlJobFileAppender.makeLogFileName(new Date(triggerParam.getLogDateTim()),
							triggerParam.getLogId());
					XxlJobFileAppender.contextHolder.set(logFileName);

					// 写入分片信息， 将当前机器的分片标记和分片总数写入到ShardingUtil中，到时候，可以在handler中通过这个工具类获取
					ShardingUtil.setShardingVo(new ShardingUtil.ShardingVO(triggerParam.getBroadcastIndex(),
							triggerParam.getBroadcastTotal()));

					// execute
					XxlJobLogger.log("<br>----------- xxl-job job execute start -----------<br>----------- Param:"
							+ triggerParam.getExecutorParams());

					// 超时时间大于0
					if (triggerParam.getExecutorTimeout() > 0) {
						// limit timeout
						Thread futureThread = null;
						try {
							final TriggerParam triggerParamTmp = triggerParam;

							// 开一个线程执行
							FutureTask<ReturnT<String>> futureTask = new FutureTask<ReturnT<String>>(
									new Callable<ReturnT<String>>() {
										@Override
										public ReturnT<String> call() throws Exception {
											// 执行。。。
											return handler.execute(triggerParamTmp.getExecutorParams());
										}
									});
							futureThread = new Thread(futureTask);
							futureThread.start();
							// 获取执行结果
							executeResult = futureTask.get(triggerParam.getExecutorTimeout(), TimeUnit.SECONDS);

						} catch (TimeoutException e) {

							XxlJobLogger.log("<br>----------- xxl-job job execute timeout");
							XxlJobLogger.log(e);

							executeResult = new ReturnT<String>(IJobHandler.FAIL_TIMEOUT.getCode(),
									"job execute timeout ");
						} finally {
							futureThread.interrupt();
						}
					} else {// 否则直接执行
						// just execute
						executeResult = handler.execute(triggerParam.getExecutorParams());
					}

					if (executeResult == null) {
						executeResult = IJobHandler.FAIL;
					}
					XxlJobLogger
							.log("<br>----------- xxl-job job execute end(finish) -----------<br>----------- ReturnT:"
									+ executeResult);
				} else {
					if (idleTimes > 30) {
						// 每3秒获取一次数据，获取30次都没有获取到数据之后，则现场被清除
						XxlJobExecutor.removeJobThread(jobId, "excutor idel times over limit.");
					}
				}
			} catch (Throwable e) {
				if (toStop) {
					XxlJobLogger.log("<br>----------- JobThread toStop, stopReason:" + stopReason);
				}

				StringWriter stringWriter = new StringWriter();
				e.printStackTrace(new PrintWriter(stringWriter));
				String errorMsg = stringWriter.toString();
				executeResult = new ReturnT<String>(ReturnT.FAIL_CODE, errorMsg);

				XxlJobLogger.log("<br>----------- JobThread Exception:" + errorMsg
						+ "<br>----------- xxl-job job execute end(error) -----------");
			} finally {
				if (triggerParam != null) {
					// callback handler info
					if (!toStop) {
						// handler执行完成之后，将结果写入到日志里面去，
						// 就是在执行器启动的时候，会建立一个线程，用来实时处理日志，此处是将结果和logID放入到队列里面去，
						TriggerCallbackThread.pushCallBack(new HandleCallbackParam(triggerParam.getLogId(),
								triggerParam.getLogDateTim(), executeResult));
					} else {
						// is killed
						ReturnT<String> stopResult = new ReturnT<String>(ReturnT.FAIL_CODE,
								stopReason + " [job running，killed]");
						// 任务执行后，放入回调队列
						TriggerCallbackThread.pushCallBack(new HandleCallbackParam(triggerParam.getLogId(),
								triggerParam.getLogDateTim(), stopResult));
					}
				}
			}
		}

		// callback trigger request in queue
		// 当现场被终止之后，队列里面剩余的未执行的任务，将被终止的这些任务放入队列，供日志监控线程来处理，回调给调度中心
		while (triggerQueue != null && triggerQueue.size() > 0) {
			TriggerParam triggerParam = triggerQueue.poll();
			if (triggerParam != null) {
				// is killed
				ReturnT<String> stopResult = new ReturnT<String>(ReturnT.FAIL_CODE,
						stopReason + " [job not executed, in the job queue, killed.]");
				// 任务执行后，放入回调队列
				TriggerCallbackThread.pushCallBack(
						new HandleCallbackParam(triggerParam.getLogId(), triggerParam.getLogDateTim(), stopResult));
			}
		}

		// destroy
		try {
			handler.destroy();
		} catch (Throwable e) {
			logger.error(e.getMessage(), e);
		}

		logger.info(">>>>>>>>>>> xxl-job JobThread stoped, hashCode:{}", Thread.currentThread());
	}
}
